package regular;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.Iterator;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public class EvalReducer extends
		Reducer<IntWritable, RvecWritable, NullWritable, RvecWritable> {

	public MultipleOutputs multipleOutputs;

	public void setup(Context context) {
		multipleOutputs = new MultipleOutputs(context);
	}

	protected void cleanup(Context context) throws IOException,
			InterruptedException {
		multipleOutputs.close();
	}

	public void reduce(IntWritable key, Iterable<RvecWritable> values,
			Context context) throws IOException, InterruptedException {

		Initial.CONF = context.getConfiguration();
		
		// get query
		Path[] fileCaches = DistributedCache.getLocalCacheFiles(context
				.getConfiguration());
		GraphQuery gq = Ultility.getAutomaton(fileCaches[0]);

		ArrayList<Rvec> RVset = new ArrayList<Rvec>();
		Iterator<RvecWritable> itr = values.iterator();
		RvecWritable data = new RvecWritable();
		while (itr.hasNext()) {
			data = itr.next();
			if (data.rvset.size() > 0) {
				RVset.addAll(data.rvset);
				
				multipleOutputs.write(NullWritable.get(), data, key.toString());
			}
		}

//		RvecWritable total = new RvecWritable();		
//		total.rvset = RVset;		
//		context.write(NullWritable.get(),total);
		
		ArrayList<String> listOut = new ArrayList<String>();
		StringBuilder sbIn = new StringBuilder();
		StringBuilder sb = new StringBuilder();
		for (Rvec r : RVset) {
			for (BoolFormula b : r.value) {
				if (!listOut.contains(b.toString())) {
					if (!b.value) {
						listOut.add(b.toString());
						sb.append(b.toString() + "\n");
					} else {
						if (!listOut.contains(gq.ut.stateName)) {
							listOut.add(gq.ut.stateName);
							sb.append(gq.ut.stateName + "\n");
						}
					}
				}
			}
			sbIn.append(r.v + "\n");
		}

		FileSystem fs = FileSystem.get(Initial.CONF);
		Path pathOutputNode = new Path(Initial.PATH_OUTPUT_NODE_CACHE + "/"
				+ key);
//		FSDataOutputStream fout = fs.create(pathOutputNode, true);
//		fout.writeUTF(sb.toString());
//		fout.close();
		BufferedWriter bwOut = new BufferedWriter(new OutputStreamWriter(fs.create(pathOutputNode, true)));
		bwOut.write(sb.toString());
		bwOut.close();

		Path pathInputNode = new Path(Initial.PATH_INPUT_NODE_CACHE + "/" + key);
//		FSDataOutputStream fin = fs.create(pathInputNode, true);
//		fin.writeUTF(sbIn.toString());
//		fin.close();
		BufferedWriter bwIn = new BufferedWriter(new OutputStreamWriter(
				fs.create(pathInputNode, true)));
		bwIn.write(sbIn.toString());
		bwIn.close();
	}
}
